package zy.learn.demo.structuredstreaming.join

import java.sql.Timestamp

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.{OutputMode, Trigger}

object StreamJoinWatermark {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().set("spark.sql.shuffle.partitions", "3")
    val spark = SparkSession.builder()
      .master("local[2]")
      .config(sparkConf)
      .appName("StreamingJoinWithWatermark")
      .getOrCreate()

    import spark.implicits._

    // 第 1 个 stream
    val nameSexStream = spark.readStream
      .format("socket")
      .option("host", "co7-203")
      .option("port", 9999)
      .load
      .as[String]
      .map(line => {
        val arr: Array[String] = line.split(",")
        (arr(0), arr(1), Timestamp.valueOf(arr(2)))
      }).toDF("name", "sex", "ts1")

    // 第 2 个 stream
    val nameAgeStream = spark.readStream
      .format("socket")
      .option("host", "co7-203")
      .option("port", 8888)
      .load
      .as[String]
      .map(line => {
        val arr: Array[String] = line.split(",")
        (arr(0), arr(1).toInt, Timestamp.valueOf(arr(2)))
      }).toDF("name", "age", "ts2")

    // join 操作
    val joinResult = nameSexStream.join(nameAgeStream, "name")
    joinResult.writeStream
      .outputMode(OutputMode.Append())
      .format("console")
      .trigger(Trigger.ProcessingTime(0))
      .start()
      .awaitTermination()
  }
}
